package com.github.ompc.carrying.client.persistence.impl;

import static com.github.ompc.carrying.client.common.CarryingClientConstants.CARRYING_CLIENT_BUFFER_BOUNDARY_VALUE;
import static com.github.ompc.carrying.client.common.CarryingClientConstants.CARRYING_CLIENT_BUFFER_SIZE;
import static com.github.ompc.carrying.client.common.CarryingClientConstants.NEW_LINE;
import static java.lang.String.format;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.ompc.carrying.client.array.DataConsumerArray;
import com.github.ompc.carrying.client.array.DataConsumerArrayManager;
import com.github.ompc.carrying.client.persistence.CarryingDataPersistenceDao;
import com.github.ompc.carrying.common.domain.Row;

public class MappedCarryingDataPersistenceDao implements CarryingDataPersistenceDao{

	private final Logger logger = LoggerFactory.getLogger(getClass());
	
	private FileChannel fileChannel;
	private MappedByteBuffer mappedBuffer;
	private File file;
	private String dataFilePath;
	private CountDownLatch latch = new CountDownLatch(1);
	
	private Runnable dataPersistenceRunnable = new Runnable(){

		@Override
		public void run() {
			
			List<DataConsumerArray> arrayingDataArrays = DataConsumerArrayManager.getDataConsumerArrays();
		
			for(int i=0;i<arrayingDataArrays.size();i++){
				Row[] rows = arrayingDataArrays.get(i).getRows();
				for(int j=0;j<rows.length;j++){
					Row row = rows[j];
					if(row==null){
						break;
					}
					mappedBuffer.put(Long.valueOf(row.getLineNum()).byteValue());
					//换行符字节表示方式
					mappedBuffer.put(rows[j].getData()).put(NEW_LINE);
					if(isEOF()){
						try {
							mappedBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 
									mappedBuffer.position(),CARRYING_CLIENT_BUFFER_SIZE);
						} catch (IOException e) {
							logger.error("error@MappedCarryingDataPersistenceDao dataPersistenceRunnable IOException",e);
						}
					}
				}
			}
			
			closeFileChannel();
			
			latch.countDown();
			
		}
	};
	
	/**
	 * 判断是否到达缓冲区末尾
	 * @return
	 */
	private boolean isEOF(){
		return mappedBuffer.remaining()<CARRYING_CLIENT_BUFFER_BOUNDARY_VALUE;
	}
	
	@Override
	public void persistenceData() throws Exception{
		
		file = new File(dataFilePath);
		
		if(null==file
				  ||!file.exists()
				  ||!file.canWrite()){
			throw new IllegalAccessException(format("file=%s access failed.", file));
		}
		fileChannel = new RandomAccessFile(file,"rw").getChannel();
		mappedBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE,0,CARRYING_CLIENT_BUFFER_SIZE);
		
		final Thread thread = new Thread(dataPersistenceRunnable,"data-persitence-thread");
		thread.start();
		
		latch.await();
		
	}
	
	/**
	 * 关闭文件通道
	 */
	private void closeFileChannel(){
		if(null != fileChannel){
			try{
				fileChannel.close();
			}catch(Throwable t){
				logger.info("info@MappedCarryingDataPersistenceDao closeFileChannel,file={}",file);
			}
		}
	}
	
	public void setFile(File file) {
		this.file = file;
	}
	
	public void setDataFilePath(String dataFilePath) {
		this.dataFilePath = dataFilePath;
	}

}
